package amqplib

import (
	"fmt"
	"time"

	"github.com/brotherbaby/jinzhugo/logger"
	"github.com/streadway/amqp"
)

func (r *RabbitMQ) formatConnUrl(cf *MQConnectorConfig) string {
	connDSN := fmt.Sprintf("amqp://%s:%s@%s:%d/%s", cf.User, cf.Password, cf.Host, cf.Port)
	logger.Info(fmt.Sprintf("amqp://%s:***@%s:%d/%s", cf.User, cf.Password, cf.Host, cf.Port))
	return connDSN
}

func (r *RabbitMQ) InitRabbitMQ(connConfig *MQConnectorConfig, poolConfig *ConnectionPoolConfig) error {
	r.config = connConfig
	r.channelMap = make(map[string]*amqp.Channel)

	pool, err := InitConnectionPool(*connConfig, *poolConfig)
	if err != nil {
		logger.Error("Failed to created RabbitMQ pool: %v", err)
	}
	r.pool = pool

	return nil
}

// 获取通道 (带缓存)
func (r *RabbitMQ) getChannel(key string) (*amqp.Channel, error) {
	r.mu.RLock()
	ch, exists := r.channelMap[key]
	r.mu.RUnlock()

	if exists && ch != nil {
		return ch, nil
	}

	// 没有缓存或已关闭，创建新通道
	conn, err := r.pool.GetConn()
	if err != nil {
		return nil, err
	}
	defer r.pool.Release(conn)

	ch, err = conn.Channel()
	if err != nil {
		return nil, err
	}

	r.mu.Lock()
	r.channelMap[key] = ch
	r.mu.Unlock()

	return ch, nil
}

// 关闭所有通道
func (r *RabbitMQ) Close() {
	r.mu.Lock()
	defer r.mu.Unlock()

	for key, ch := range r.channelMap {
		if ch != nil {
			ch.Close()
		}
		delete(r.channelMap, key)
	}

	if r.pool != nil {
		r.pool.Close()
	}
}

func (r *RabbitMQ) Publish(pm MQPublishMessage, rpcMode bool) error {
	exchangeName := pm.Exchange
	// channelMap的key自己生成，为了可复用不重复过多创建channel
	ch, err := r.getChannel("publish-" + exchangeName + "-" + pm.RoutingKey)
	if err != nil {
		return err
	}

	// 确保交换机存在
	if pm.Exchange != "" {
		err = ch.ExchangeDeclare(
			exchangeName,    // name
			pm.ExchangeType, // type
			pm.Durable,      // durable
			false,           // auto-deleted
			false,           // internal
			false,           // no-wait
			nil,             // arguments
		)
		if err != nil {
			return err
		}
	}

	if pm.SkipExchange {
		exchangeName = ""
	}

	if rpcMode {
		// 声明 replyTo 队列
		_, err = ch.QueueDeclare(
			pm.ReplyTo, // name
			false,      // durable
			true,       // delete when unused
			false,      // exclusive
			false,      // no-wait
			nil,        // arguments
		)
		if err != nil {
			return err
		}

		// 启动一个 goroutine 监听 replyTo 队列
		go func() {
			msgs, err := ch.Consume(
				pm.ReplyTo, // queue
				"",         // consumer
				true,       // auto-ack
				false,      // exclusive
				false,      // no-local
				false,      // no-wait
				nil,        // args
			)
			if err != nil {
				logger.Error("Failed to consume from replyTo queue: %v", err)
				return
			}

			for d := range msgs {
				if d.CorrelationId == pm.CorrelationID {
					response := MQConsumerMessage{
						Body:          d.Body,
						CorrelationID: d.CorrelationId,
						ReplyTo:       d.ReplyTo,
					}
					pm.Response <- response
					break
				}
			}
		}()
	}

	// 发布消息
	return ch.Publish(
		exchangeName,  // exchange
		pm.RoutingKey, // routing key, it's queue name while exchange=''
		false,         // mandatory
		false,         // immediate
		amqp.Publishing{
			ContentType:   "application/json",
			Body:          pm.Body,
			Timestamp:     time.Now(),
			DeliveryMode:  amqp.Transient,
			CorrelationId: pm.CorrelationID,
			ReplyTo:       pm.ReplyTo,
		})
}

func (r *RabbitMQ) Consume(cm MQConsumerMessage) error {
	ch, err := r.getChannel("consume-" + cm.Queue)
	if err != nil {
		return err
	}

	// 确保队列存在
	_, err = ch.QueueDeclare(
		cm.Queue,   // name
		cm.Durable, // durable
		false,      // delete when unused
		false,      // exclusive
		false,      // no-wait
		nil,        // arguments
	)
	if err != nil {
		return err
	}

	// 开始消费
	msgs, err := ch.Consume(
		cm.Queue,       // queue
		cm.ConsumerTag, // consumer
		cm.AutoAck,     // auto-ack
		false,          // exclusive
		false,          // no-local
		false,          // no-wait
		nil,            // args
	)
	if err != nil {
		return err
	}

	if nil != cm.Callback {
		// 异步处理消息
		go func() {
			for d := range msgs {
				if "" != d.CorrelationId && "" != d.ReplyTo {
					responseMsg := cm.Callback(d)
					if responseMsg != nil {
						// 发送响应到 replyTo 队列
						responseCh, err := r.getChannel("publish-" + responseMsg.Exchange + "-" + responseMsg.RoutingKey)
						if err != nil {
							logger.Error("Failed to get channel for sending response: %v", err)
							continue
						}

						err = responseCh.Publish(
							responseMsg.Exchange, // exchange
							d.ReplyTo,            // routing key (replyTo queue)
							false,                // mandatory
							false,                // immediate
							amqp.Publishing{
								ContentType:   "application/json",
								Body:          responseMsg.Body,
								Timestamp:     time.Now(),
								DeliveryMode:  amqp.Transient,
								CorrelationId: d.CorrelationId,
							})
						if err != nil {
							logger.Error("Failed to send response to replyTo queue: %v", err)
						}
					}
				} else {
					cm.Callback(d)
				}
				if !cm.AutoAck {
					d.Ack(false)
				}
			}
		}()
	}

	return nil
}
